package org.timewindow

import org.FlinkStreamApp
import org.apache.flink.api.common.state.ValueStateDescriptor
import org.apache.flink.api.scala.typeutils.Types
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector
import org.bean.SensorReading
import org.diysource.SensorSource

/**
 * description ：传感器实现一分钟内连续温度上升预警
 * author      ：剧情再美终是戏
 * mail        : 13286520398@163.com
 * date        ：Created in 2020/2/21 20:41
 * modified By ：
 * version:    : 1.0
 */
object TempProcessFunctionExample extends FlinkStreamApp {

  override def doSomeThing(environment: StreamExecutionEnvironment) = {
    // 创建数据
    val source = environment.addSource(new SensorSource)
    //source.print()

    // process
    val warings = source.keyBy(_.id).process(new TempAlterFunction())
    warings.print()
  }

  class TempAlterFunction extends KeyedProcessFunction[String, SensorReading, String] {
    // 记录上一次温度 默认值是0.0
    lazy val lastTemp = getRuntimeContext.getState(
      new ValueStateDescriptor[Double]("lastTemp", Types.of[Double])
    )

    // 记录定时器 默认值是 0
    lazy val timer = getRuntimeContext.getState(
      new ValueStateDescriptor[Long]("timer", Types.of[Long])
    )

    // 获取数据，逻辑处理
    override def processElement(value: SensorReading,
                                ctx: KeyedProcessFunction[String, SensorReading, String]#Context,
                                out: Collector[String]) = {
      // 获取状态值
      val cureTemp = lastTemp.value()
      val cureTimer = timer.value()
      // 更新本次的温度
      val temperature = value.temperature
      lastTemp.update(temperature)

      if (cureTimer == 0 && temperature > cureTemp) { // 如果定时器未初始化 && 传感器温度上涨了
        // 注册定时器
        val timerTs = ctx.timerService().currentProcessingTime() + 1000 // 这里可以设置定时器时间
        ctx.timerService().registerProcessingTimeTimer(timerTs)

        // 将定时器时间保存
        timer.update(timerTs)
      } else if (cureTemp == 0.0 || temperature < cureTemp) { // 如果上一次温度为 0.0 || 温度下降了
        // 删除定时器
        ctx.timerService().deleteProcessingTimeTimer(cureTimer)
        timer.clear()
      }
      // 剩下的就是温度上涨 && 定时器已经被初始化，到了定时器触发时就会调用 onTimer
    }

    // 预警写出
    override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[String, SensorReading, String]#OnTimerContext, out: Collector[String]) = {
      // 预警写出
      out.collect("传感器id为： " + ctx.getCurrentKey + "的传感器温度值已经连续 1s 上升了。")

      // 清空状态变量
      timer.clear()
    }
  }

}
